import com.nekonomics.lakehouse.common.FlinkRunner._
import com.nekonomics.lakehouse.common.util.SqlUtil._
import com.nekonomics.lakehouse.common.adt.PaymentStatus._

/**
 * 商家商品销售排名
 * 商家、spu为粒度
 * topN为统计指标
 * 统计周期为天
 */

object MerchantSpuTopnDailyAds{
  def main(args: Array[String]): Unit = {
    implicit val config: SqlRunnerConfig = SqlRunnerConfig(groupId = "hot_product_rank_daily", port = 8086, ckp = true)
    runSql { (_, tableEnv) =>
      tableEnv.executeSql(
        sql"""
             |CREATE TEMPORARY TABLE  merchant_spu_topn_daily_ads (
             |  date_key DATE,
             |  merchant_id BIGINT,
             |  spu_id BIGINT,
             |  product_name STRING,
             |  product_quantity INT,
             |  product_sales DECIMAL(10, 2),
             |  product_rank INT,
             |  PRIMARY KEY (date_key, merchant_id, product_rank) NOT ENFORCED
             |) WITH ${jdbcConnector("ads.merchant_spu_topn_daily_ads")}
             |""".stripMargin
      )


      tableEnv.executeSql(
        sql"""
             |INSERT INTO merchant_spu_topn_daily_ads
             |SELECT
             | date_key,
             | merchant_id,
             | spu_id,
             | product_name,
             | product_quantity,
             | product_sales,
             | CAST(product_rank AS INT) AS product_rank
             |FROM (
             |  SELECT
             |      date_key,
             |      merchant_id,
             |      spu_id,
             |      MAX(product_name) AS product_name,
             |      SUM(total_quantity) AS product_quantity,
             |      SUM(total_sales) AS product_sales,
             |      ROW_NUMBER() OVER (PARTITION BY merchant_id, date_key ORDER BY SUM(total_quantity) DESC)  AS product_rank
             |  FROM dws.order_sku_sale_daily_dws
             |  GROUP BY
             |      date_key,
             |      merchant_id,
             |      spu_id
             |) t
             |WHERE product_rank <= 10;
             |""".stripMargin
      )


    }

  }


}
